package com.delay.queue.delayedqueue.redissortedset;

import com.alibaba.fastjson.JSON;
import com.delay.queue.model.MessageTask;
import org.apache.commons.collections.CollectionUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author gxing
 * @desc Sorted Set 实现延时队列
 * @date 2021/3/9
 */
@Component
public class DelayQueueManager implements CommandLineRunner {
    private static final Logger logger = LogManager.getLogger(DelayQueueManager.class);

    private static final String DELAY_MESSAGE_QUEUE_NAME = "MessageQueue:Task";

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public void run(String... args) throws Exception {
        ValueOperations<String, Object> operations = redisTemplate.opsForValue();
        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(this::executeTask);
    }

    /**
     * 延时任务执行线程
     */
    private void executeTask() {
        while (true) {
/*            // 这里演示用 while(true), 生产最好使用定时任务
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 方式一, 基于 rangeByScoreWithScores 方法
            long currentSecond = (System.currentTimeMillis() / 1000);
            Set<ZSetOperations.TypedTuple<Object>> typedTuples = redisTemplate.opsForZSet().rangeByScoreWithScores(DELAY_MESSAGE_QUEUE_NAME, 0, currentSecond, 0, 100);
            if (CollectionUtils.isEmpty(typedTuples)) {
                logger.info("-------------延时队列为空-------------");
                continue;
            }
            for (ZSetOperations.TypedTuple<Object> typedTuple : typedTuples) {
                MessageTask messageTask = JSON.parseObject(JSON.toJSONString(typedTuple.getValue()), MessageTask.class);
                // 防止并发问题,执行先删除,再处理业务,删除成功返回1,并发已被删除的任务会删除失败就不执行任务
                long result = redisTemplate.opsForZSet().remove("MessageQueue:Task", messageTask);
                if (result == 1) {
                    System.out.println("处理业务：" + JSON.toJSONString(messageTask));
                }
            }*/


            // 方式二,基于 rangeWithScores 方法,但在业务层需要判断是否过期
/*            Set<ZSetOperations.TypedTuple<Object>> typedTuples = redisTemplate.opsForZSet().rangeWithScores(DELAY_MESSAGE_QUEUE_NAME, 0, 100);
            if (CollectionUtils.isEmpty(typedTuples)) {
                logger.info("-------------延时队列为空-------------");
                continue;
            }

            for (ZSetOperations.TypedTuple<Object> typedTuple : typedTuples) {
                long score = typedTuple.getScore().longValue();
                long currentSecond = (System.currentTimeMillis() / 1000);
                logger.info("score:{}, current:{}", score, currentSecond);
                MessageTask messageTask = JSON.parseObject(JSON.toJSONString(typedTuple.getValue()), MessageTask.class);
                if (score < currentSecond) {
                    // 防止并发问题,执行先删除,再处理业务,删除成功返回1,并发已被删除的任务会删除失败就不执行任务
                    long result = redisTemplate.opsForZSet().remove("MessageQueue:Task", messageTask);
                    if (result == 1) {
                        System.out.println("处理业务：" + JSON.toJSONString(messageTask));
                    }
                } else {
                    // 未过期,直接跳出循环,此元素未过期,后续其它元素肯定未过期
                    break;
                }
            }*/
        }
    }

}
